package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"os/signal"
	"syscall"

	"github.com/go-mysql-org/go-mysql/mysql"
	"github.com/go-mysql-org/go-mysql/replication"
)

func main() {
	// 创建一个 Binlog Syncer 实例
	cfg := replication.BinlogSyncerConfig{
		ServerID: 1, // 唯一的 Server ID
		Flavor:   "mysql",
		Host:     "127.0.0.1",
		Port:     3307,
		User:     "repl",
		Password: "example",
	}

	syncer := replication.NewBinlogSyncer(cfg)

	// 获取当前的 Binlog 文件名和位置
	// 你可以从数据库中查询 `SHOW MASTER STATUS` 来获取当前的 Binlog 位置
	// 这里我们假设从最新的位置开始
	streamer, err := syncer.StartSync(mysql.Position{Name: "", Pos: 4})
	if err != nil {
		log.Fatalf("Failed to start sync: %v", err)
	}

	// 创建一个上下文用于优雅地关闭
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// 监听系统信号，以便优雅地关闭
	signals := make(chan os.Signal, 1)
	signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM)

	go func() {
		<-signals
		cancel()
	}()

	// 处理 Binlog 事件
	for {
		select {
		case <-ctx.Done():
			log.Println("Exiting...")
			return
		default:
			ev, err := streamer.GetEvent(ctx)
			if err != nil {
				log.Fatalf("Failed to get event: %v", err)
			}

			// 处理事件
			handleEvent(ev)
		}
	}
}

func handleEvent(ev *replication.BinlogEvent) {
	switch ev.Header.EventType {
	case replication.ROTATE_EVENT:
		fmt.Println("新的binlog文件已生成" + replication.ROTATE_EVENT.String())
	case replication.QUERY_EVENT:
		event := ev.Event.(*replication.QueryEvent)
		if string(event.Schema) == "demo" {
			fmt.Printf("query event: schema %s, query: %s \n", event.Schema, event.Query)
		}
	case replication.WRITE_ROWS_EVENTv2, replication.WRITE_ROWS_EVENTv1:
		event := ev.Event.(*replication.RowsEvent)
		if string(event.Table.Schema) == "demo" {
			fmt.Printf("insert event: table %s, rows: %v \n", event.Table.Table, event.Rows)
		}
	case replication.UPDATE_ROWS_EVENTv2, replication.UPDATE_ROWS_EVENTv1:
		fmt.Println("Update event")
	case replication.DELETE_ROWS_EVENTv2, replication.DELETE_ROWS_EVENTv1:
		fmt.Println("Delete event")
	}
}
